package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

public class DosHurstBolt extends AbstractBolt {


	private static final long serialVersionUID = 1L;			
	private HashMap<String,Double> tripleMap = new HashMap<String,Double>();
	private HashMap<String,Double> tripleOctMap = new HashMap<String,Double>();	
	private DosHurstSliding cache;

	public DosHurstBolt(){		
		cache = new DosHurstSliding(60);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","hurst"));	
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{	
			Logger.getRootLogger().info("=====tuple====="+tuple);
			String dip = tuple.getStringByField("dip").trim();
			
			//目的IP的包数和
			String tempuppkg = tuple.getStringByField("uppacket").trim();		
			double uppacket =  Double.parseDouble(tempuppkg);		
			if(tripleMap.containsKey(dip)){				
				double tmpsum = tripleMap.get(dip);				
				double triplePV = tmpsum + uppacket;
				tripleMap.put(dip, triplePV);
			}else{
				tripleMap.put(dip, uppacket);
			}			
			//目的IP的上行流量和
			String tempupoct = tuple.getStringByField("upoct").trim();
			double upoct =  Double.parseDouble(tempupoct);
			if(tripleOctMap.containsKey(dip)){				
				double tmpoctsum = tripleOctMap.get(dip);				
				double tripleOct = tmpoctsum + upoct;
				tripleOctMap.put(dip, tripleOct);
			}else{
				tripleOctMap.put(dip, upoct);
			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {
		try{
			
		Logger.getRootLogger().info("======emitCountingData start========");
		///********************用上行流包数UpStreamPacket计算*********************************/
//		if(tripleMap.size()==0){
//			Logger.getRootLogger().info(" >>>>>>>>>> tripleMap is null");
//			return;
//		}		
//		Logger.getRootLogger().info(" >>>>>>>>>> tripleMap.size()"+tripleMap.size());
//		cache.addPInfo(tripleMap);
		
		///********************用上行流字节数,整个以太网包UpStreamOctets计算*********************************/
		if(tripleOctMap.size()==0){
			Logger.getRootLogger().info(" >>>>>>>>>> tripleOctMap is null");
			return;
		}		
		Logger.getRootLogger().info(" >>>>>>>>>> tripleOctMap.size()"+tripleOctMap.size());
		cache.addPInfo(tripleOctMap);
		
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);		
		HashMap<String,String> dipHurst = new HashMap<String,String>();
		Logger.getRootLogger().info("loop start-----infoMap.size()--"+infoMap.size());	
		
//		//循环30个滑动窗口
		for(HashMap<String,Double> singleMap: infoMap.values()) {
			if(singleMap == null) continue;
			for (Entry<String, Double> entry : singleMap.entrySet()) {
				if(dipHurst.containsKey(entry.getKey())){
					String temppkg = dipHurst.get(entry.getKey());					
					temppkg = temppkg + entry.getValue()+"|";					
					dipHurst.put(entry.getKey(), temppkg);
			
				}else{					
					if(!entry.getKey().isEmpty())
						dipHurst.put(entry.getKey(), entry.getValue()+"|");
					Logger.getRootLogger().info(">>>>>>>>>>:"+entry.getKey()+">>>"+entry.getValue());	
				}
			}
		}
		
		for (Entry<String, String> hurstent : dipHurst.entrySet()) {			
			String[] arr = hurstent.getValue().split("\\|");				
			if(arr.length<5) continue;
			double hurstdata[] = new double[arr.length];
			for(int i=0;i<arr.length;i++){				
				if(arr[i].trim().isEmpty()) continue;					
				hurstdata[i] = Double.parseDouble(arr[i].trim());					
			}
			Logger.getRootLogger().info(">>>>>>>>>>5555555:" + hurstent.getKey() + ">>>>>>>>>>" + hurstent.getValue());	
			com.aotain.common.Hurst H = new com.aotain.common.Hurst(hurstdata);
			double ret = H.Calculate();
			if(Double.isNaN(ret)) continue;
			
			Logger.getRootLogger().info(">>>>key" + hurstent.getKey() + ">>>>value>" +hurstent.getValue()+">>>hurst:"+ ret);
			String accesstime = new SimpleDateFormat("yyyyMMddHHmm").format(new Date(System.currentTimeMillis()+15000));
//			String rowkey = String.format("%s_%s_pkg", hurstent.getKey(),accesstime);
			String rowkey = String.format("%s_%s_test", hurstent.getKey(),accesstime);
			String tbName = "TEMP_DDOSHURST";			
	    	hbaseInstance.Add(tbName, rowkey, "cf", "ACCESSTIME", accesstime);
	    	hbaseInstance.Add(tbName, rowkey, "cf", "DIP", hurstent.getKey());
	    	hbaseInstance.Add(tbName, rowkey, "cf", "HURST",String.valueOf(ret));
			collector.emit(new Values(hurstent.getKey(),ret));
		
		}
		
		infoMap.clear();
		tripleOctMap.clear();
		cache.advanceHead();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			
			throw new FailedException("DosHurstBolt throws an exception!",e);
		}	
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			//if(isTickTuple(tuple)) {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("DosHurstBolt throws an exception!");
		}	
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
	}
	

}
